package io.github.wanggit.antrpc.client.monitor.report;

import io.github.wanggit.antrpc.commons.bean.RpcCallLog;
import io.github.wanggit.antrpc.commons.config.IConfiguration;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;

public class KafkaLogReporter extends AbstractLogReporter {

    @SuppressWarnings("rawtypes")
    private final KafkaTemplate kafkaTemplate;

    private final String topic;

    public KafkaLogReporter(IConfiguration configuration, ApplicationContext context) {
        super(configuration);
        kafkaTemplate = context.getBean(KafkaTemplate.class);
        topic = configuration.getCallLogReporterConfig().getKafkaTopic();
    }

    @Override
    protected void internalReport(
            RpcCallLog log, ZipkinAllMethodTracing.TracingPairs lastTracingPairs) {
        if (null != kafkaTemplate) {
            //noinspection unchecked
            kafkaTemplate.send(topic, JsonUtils.toJsonString(log));
        }
    }
}
